package com.wzl.consumer.config.kafkaConfig;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

/**
 * kafka消费者消息过滤器
 *
 * @author weizhongli
 * @since 2022-09-14 17:31
 */
@Component
public class ConsumerMessageFilter {

    @Autowired
    ConsumerFactory consumerFactory;

    // 消息过滤器
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {

        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();

        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        // 消息过滤策略  根据实际情况编写过滤条件
        factory.setRecordFilterStrategy(consumerRecord -> {
            if (String.valueOf(consumerRecord.value()).contains("王")) {
                return false;
            }
            //返回true消息则被过滤
            return true;
        });
        return factory;
    }

}